package action

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object E2_Action_topN {

  def main(args: Array[String]): Unit = {
    topN()
  }
  """
    |（1）统计 Top10 热门品类中每个品类的 Top10 活跃 Session
    |鞋 点击数 下单数 支付数
    |衣服 点击数 下单数 支付数
    |电脑 点击数 下单数 支付数
    |
    |""".stripMargin
  def topN() = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("action")
    val sc = new SparkContext(sparkConf)
    val textRdd = sc.textFile("file:///D:\\workspace\\lab\\learnbigdata\\learnspark\\doc\\spark-sql\\data\\user_visit_action.txt")
    // 缓存数据
    textRdd.cache()
    // 将数据转换结构
    //   点击的场合 : ( 品类ID，( 1, 0, 0 ) )
    //   下单的场合 : ( 品类ID，( 0, 1, 0 ) )
    //   支付的场合 : ( 品类ID，( 0, 0, 1 ) )
    val flatRDD: RDD[(String, (Int, Int, Int))] = textRdd.flatMap(action => {
      val datas = action.split(",")
      if (datas(6) != "-1") {
        // 点击场合
        List((datas(6), (1, 0, 0)))
      } else if (datas(8) != "null") {
        // 下单的场合
        val ids = datas(8).split(",")
        ids.map(id => (id, (0, 1, 0)))
      } else if (datas(10) != "null") {
        // 支付的场合
        val ids = datas(10).split(",")
        ids.map(id => (id, (0, 0, 1)))
      } else {
        Nil
      }
    })
    // 将相同的品类ID的数据进行分组聚合
    //  (品类ID，(点击数量, 下单数量, 支付数量))
    val analysisRDD: RDD[(String, (Int, Int, Int))] = flatRDD.reduceByKey((t1, t2) => {
      (t1._1 + t2._1, t1._2 + t2._2, t1._3 + t2._3)
    })
    // 按照点击数、下单数、支付数
    val sortedCategories = analysisRDD.sortBy(_._2, false)
    //    val sortedCategories = analysisRDD.sortBy(_._2._1, false)
    //    val sortedCategories = analysisRDD.sortBy(_._2._2, false)
    //    val sortedCategories = analysisRDD.sortBy(_._2._3, false)
    val top10Categories = sortedCategories.take(10)
    // 获取 top10 的 品类id
    val top10Ids = top10Categories.map(_._1)

    val filterActionRdd = textRdd.filter(action => {
      val datas = action.split(",")
      if (datas(6) != "-1") {
        top10Ids.contains(datas(6))
      } else {
        false
      }
    })
    // 根据品类ID和sessionid进行点击量的统计
    // ((品类ID，sessionId）,sum)
    val reduceRdd = filterActionRdd.map(
      action => {
        val datas = action.split(",")
        ((datas(6), datas(2)), 1)
      }
    ).reduceByKey(_ + _)
    // 将统计的结果进行结构的转换
    // ((品类ID，sessionId）,sum) => (品类ID, (sessionId, sum))
    val mapRdd = reduceRdd.map {
      case ((cid, sid), num) => {
        (cid, (sid, num))
      }
    }
    // 相同的品类进行分组
    val groupRDD: RDD[(String, Iterable[(String, Int)])] = mapRdd.groupByKey()
    // 将分组后的数据进行点击量的排序，取前10名
    val resultRDD = groupRDD.mapValues(
      iter => {
        iter.toList.sortBy(_._2)(Ordering.Int.reverse).take(10)
      }
    )

    // 打印结果
//    resultRDD.collect().foreach {
//      case (id, list) =>
//        println(id)
//        list.foreach {
//          case (session, num) => {
//            println(session + "," + num)
//          }
//        }
//        println()
//    }
    resultRDD.collect().foreach(println)
    sc.stop
  }
}
